package com.azzcs.myrpc.client;

import com.azzcs.myrpc.register.ServerRegister;
import com.azzcs.myrpc.register.ZookeeperServerRegister;
import com.azzcs.myrpc.request.RpcEncoder;
import com.azzcs.myrpc.request.RpcRequest;
import com.azzcs.myrpc.serializer.Serializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @Author: wzg
 * @Date: 2021/1/22 下午5:07
 */
public class RpcClient {

    private static ExecutorService executor =
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private static ScheduledExecutorService connectionExecutor =
            Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors());

    @Autowired
    private Serializer serializer;

    @Autowired
    private ServerRegister serverRegister;

    private Map<String,ClientHandler<String>> clientHandlerMap = new HashMap<>();

    private Map<String, Set<String>> serverMap = new HashMap<>();


    public RpcClient() {
    }

    public <T> T build(String serverName,Class<T> clazz) {
        List<String> serverAddress = serverRegister.getServerAddress(serverName);
        for (String address : serverAddress) {
            connectionServer(address);
            Set<String> set = getAddress(serverName);
            set.add(address);
        }
        serverRegister.registerListener(this,serverName);
        scheduleReport(serverName);
        return (T) Proxy.newProxyInstance(this.getClass().getClassLoader(),
                new Class[]{clazz},
                new MyInvocationHandler(serverName));
    }

    private Set<String> getAddress(String serverName){
        serverMap.putIfAbsent(serverName,new HashSet<>());
        return serverMap.get(serverName);
    }
    private void scheduleReport(String serverName) {
        connectionExecutor.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                System.out.println("每5秒刷新一次过期的服务响应时间");
                serverRegister.scheduleReport(serverName);
            }
        },5,5, TimeUnit.SECONDS);
    }

    private void connectionServer(String address) {
        String[] split = address.split(":");
        String host = split[0];
        int port = Integer.parseInt(split[1]);
        ClientHandler<String> handler = new ClientHandler();
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new StringDecoder());
                        p.addLast(new RpcEncoder(RpcRequest.class,serializer));
                        p.addLast(handler);
                    }
                });
        try {
            b.connect(host, port).sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("客户端链接服务器："+address);
        clientHandlerMap.put(address,handler);
    }

    public void downServerAddress(List<String> down) {
        for (String address : down) {
            clientHandlerMap.remove(address);
            System.out.println(address + "下线");
        }
    }

    public void upServerAddress(List<String> up) {
        for (String address : up) {
            connectionServer(address);
        }
    }

    public void updateServerAddress(String serverName, List<String> address) {
        Set<String> currentServer = getAddress(serverName);
        List<String> downList = getDownList(currentServer, address);
        downServerAddress(downList);
        List<String> upList = getUpList(currentServer, address);
        upServerAddress(upList);
        // 重置服务列表
        currentServer.removeAll(downList);
        currentServer.addAll(upList);
        serverMap.put(serverName,currentServer);
    }



    private List<String> getDownList(Set<String> currentServer, List<String> address) {
        List<String> down = new ArrayList<>();
        for (String addr : currentServer) {
            if (!address.contains(addr)) {
                down.add(addr);
            }
        }
        return down;
    }

    private List<String> getUpList(Set<String> currentServer, List<String> address) {
        List<String> up = new ArrayList<>();
        for (String addr : address) {
            if (!currentServer.contains(addr)) {
                up.add(addr);
            }
        }
        return up;
    }

    private class MyInvocationHandler implements InvocationHandler {

        private String serverName;
        public MyInvocationHandler(String serverName) {
            this.serverName = serverName;
        }

        @Override
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
            String address = serverRegister.loadBalance(serverName);
            ClientHandler<String> handler = clientHandlerMap.get(address);


            Class<?> clazz = method.getDeclaringClass();
            String className = clazz.getName();
            String methodName = method.getName();
            Class<?>[] parameterTypes = method.getParameterTypes();

            RpcRequest request = new RpcRequest();
            request.setClassName(className);
            request.setMethodName(methodName);
            request.setParameterTypes(parameterTypes);
            request.setParameters(args);

            handler.setParameter(request);
            long startTime = System.currentTimeMillis();
            String result = executor.submit(handler).get();
            long endTime = System.currentTimeMillis();
            long useTime = endTime - startTime;
            System.out.println(serverName+"服务请求地址为："+address+"调用花费："+useTime+" 毫秒");
            serverRegister.reportTime(serverName,address,useTime);
            return result;
        }
    }

}
